﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using UtilZ.Dotnet.Ex.Base;
using UtilZ.Dotnet.Ex.Exceptions;
using UtilZ.Dotnet.Ex.Log;
using UtilZ.Dotnet.Ex.ZLog;

namespace UtilZ.Dotnet.Ex.DataStruct
{
    /// <summary>
    /// 异步队列
    /// </summary>
    /// <typeparam name="T">数据类型</typeparam>
    public class AsynQueue<T> : IDisposableZ
    {
        /// <summary>
        /// 队列消费者线程
        /// </summary>
        private readonly ThreadEx _thread = null;

        /// <summary>
        /// 获取队列线程状态[true:线程正在运行;false:线程未运行]
        /// </summary>
        public bool Status
        {
            get { return this._thread.IsRuning; }
        }

        /// <summary>
        /// Queue
        /// </summary>
        private readonly Queue<T> _queue;

        /// <summary>
        /// 选项
        /// </summary>
        private readonly AsynQueueOptions _options;

        /// <summary>
        /// 空队列等待线程消息通知
        /// </summary>
        private readonly AutoResetEvent _emptyQueueWaitEventHandle = new AutoResetEvent(false);

        private void EmptyQueueWaitEventHandleSet()
        {
            try
            {
                this._emptyQueueWaitEventHandle.Set();
            }
            catch (ObjectDisposedException)
            { }
        }


        /// <summary>
        /// 对象是否已释放[true:已释放;false:未释放]
        /// </summary>
        private bool _disposed = false;
        /// <summary>
        /// 获取当前对象是否已释放[true:已释放;false:未释放]
        /// </summary>
        bool IDisposableZ.Disposabled
        {
            get { return _disposed; }
        }

        /// <summary>
        /// 是否每次输出集合[true:输出集合;false:输出单项]
        /// </summary>
        private readonly bool _outputCollection;

        /// <summary>
        /// 数据处理委托
        /// </summary>
        private readonly Action<T> _processSingleItemAction;

        /// <summary>
        /// 数据处理委托[第一个参数：元素数组；第二个参数有效元素个数]
        /// </summary>
        private readonly Action<T[], int> _processCollectionAction;

        /// <summary>
        /// 同步操作对象
        /// </summary>
        public readonly object SyncRoot = new object();





        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="outputCollection">是否每次输出集合</param>
        /// <param name="options">选项</param>
        private AsynQueue(bool outputCollection, AsynQueueOptions options)
        {
            if (options.Capcity < 1)
            {
                throw new ArgumentException(nameof(options.Capcity));
            }

            this._outputCollection = outputCollection;
            this._queue = new Queue<T>();
            this._options = options;
            this._thread = new ThreadEx(this.RunThreadQueueProcessMethod, options.ThreadName, options.IsBackground);
            if (options.AutoStart)
            {
                this.Start();
            }
        }

        /// <summary>
        /// 构造函数-单项输出
        /// </summary>
        /// <param name="processAction">数据处理委托</param>
        /// <param name="options">选项</param>
        public AsynQueue(Action<T> processAction, AsynQueueOptions options) :
            this(false, options)
        {
            this._processSingleItemAction = processAction;
        }



        /// <summary>
        /// 构造函数-批量输出
        /// </summary>
        /// <param name="processAction">数据处理委托[第一个参数：元素数组；第二个参数有效元素个数]</param>
        /// <param name="options">选项</param>
        public AsynQueue(Action<T[], int> processAction, AsynQueueOptions options) :
            this(true, options)
        {
            if (options.BatchCount < 1)
            {
                throw new ArgumentException(string.Format("批量处理最大项数不能小于1,值:{0}无效", options.BatchCount));
            }

            this._processCollectionAction = processAction;
        }






        /// <summary>
        /// 启动子类无参数工作线程
        /// </summary>
        public void Start()
        {
            this.ThrowObjectDisposedException();
            this._thread.Start();
        }

        /// <summary>
        /// 停止工作线程
        /// </summary>       
        /// <param name="sync">是否同步停止[true:同步停止;false:异常停止];注:注意线程死锁,典型场景:刷新UI,在UI上执行同步停止</param>
        /// <param name="synMillisecondsTimeout">同步超时时间,-1表示无限期等待,单位/毫秒[isSycn为true时有效]</param>
        public void Stop(bool sync = false, int synMillisecondsTimeout = -1)
        {
            if (sync)
            {
                this._thread.SyncStop(synMillisecondsTimeout);
            }
            else
            {
                this._thread.Stop();
            }
            this.EmptyQueueWaitEventHandleSet();
        }

        /// <summary>
        /// 线程队列处理方法
        /// </summary>
        private void RunThreadQueueProcessMethod(ThreadExPara threadPara)
        {
            CancellationToken token = threadPara.Token;
            if (this._outputCollection)
            {
                this.RunThreadQueueMuiltProcessMethod(token);
            }
            else
            {
                this.RunThreadQueueSingleProcessMethod(token);
            }
        }

        /// <summary>
        /// 线程队列处理方法
        /// </summary>
        private void RunThreadQueueSingleProcessMethod(CancellationToken token)
        {
            try
            {
                bool result;
                T item;
                while (!token.IsCancellationRequested)
                {
                    try
                    {
                        lock (this.SyncRoot)
                        {
                            result = this._queue.TryDequeue(out item);
                        }

                        if (result)
                        {
                            //数据处理
                            this._processSingleItemAction?.Invoke(item);
                        }
                        else
                        {
                            try
                            {
                                this._emptyQueueWaitEventHandle.WaitOne(this._options.EmptyQueueWaitTimeout);
                            }
                            catch (ObjectDisposedException)
                            {
                                continue;
                            }
                        }
                    }
                    catch (Exception exi)
                    {
                        ZLog.ZLoger.Error(exi);
                    }
                }
            }
            catch (Exception ex)
            {
                ZLog.ZLoger.Error(ex);
            }
        }

        /// <summary>
        /// 线程队列处理方法
        /// </summary>
        private void RunThreadQueueMuiltProcessMethod(CancellationToken token)
        {
            try
            {
                int batchCount = this._options.BatchCount;
                T[] arr = new T[batchCount];
                int index, count;
                bool result;

                while (!token.IsCancellationRequested)
                {
                    try
                    {
                        count = 0;
                        index = 0;

                        while (count < batchCount && !token.IsCancellationRequested)
                        {
                            lock (this.SyncRoot)
                            {
                                result = this._queue.TryDequeue(out arr[index]);
                            }

                            if (result)
                            {
                                index++;
                                count++;
                            }
                            else
                            {
                                try
                                {
                                    if (this._emptyQueueWaitEventHandle.WaitOne(this._options.BatchMillisecondsTimeout))
                                    {
                                        continue;
                                    }
                                    else
                                    {
                                        break;
                                    }
                                }
                                catch (ObjectDisposedException)
                                {
                                    continue;
                                }
                            }
                        }

                        if (count > 0 && !token.IsCancellationRequested)
                        {
                            //数据处理
                            this._processCollectionAction?.Invoke(arr, count);//数据处理
                        }
                        else
                        {
                            try
                            {
                                this._emptyQueueWaitEventHandle.WaitOne(this._options.EmptyQueueWaitTimeout);
                            }
                            catch (ObjectDisposedException)
                            {
                                continue;
                            }
                        }
                    }
                    catch (Exception exi)
                    {
                        ZLog.ZLoger.Error(exi);
                    }
                }
            }
            catch (Exception ex)
            {
                ZLog.ZLoger.Error(ex);
            }
        }


        /// <summary>
        /// 将对象添加到队列的结尾处
        /// </summary>
        /// <param name="item">待添加的对象</param>
        public void Enqueue(T item)
        {
            lock (this.SyncRoot)
            {
                if (this._queue.Count >= this._options.Capcity)
                {
                    switch (this._options.FullMode)
                    {
                        case ContainerFullMode.Exception:
                            throw new ContainerFullException();
                        case ContainerFullMode.DropOld:
                            //将最后一项出队列
                            this._queue.Dequeue();
                            break;
                        case ContainerFullMode.DropNew:
                            throw new NotSupportedException(this._options.FullMode.ToString());
                        case ContainerFullMode.Ignore:
                            //直接返回即可
                            ZLoger.Warn($"{nameof(AsynQueue<T>)}忽略数据项");
                            return;
                        default:
                            throw new NotImplementedException(this._options.FullMode.ToString());
                    }
                }

                this._queue.Enqueue(item);
                this.EmptyQueueWaitEventHandleSet();
            }
        }

        /// <summary>
        /// 移除位于开始处的指定个数对象
        /// </summary>
        /// <param name="count">要移除的项数</param>
        /// <returns>移除项集合</returns>
        public List<T> Remove(int count)
        {
            var removeItems = new List<T>();
            T item;
            lock (this.SyncRoot)
            {
                while (removeItems.Count < count && this._queue.Count > 0)
                {
                    item = this._queue.Dequeue();
                    removeItems.Add(item);
                }
            }

            return removeItems;
        }

        /// <summary>
        /// 移除满足条件的元素
        /// </summary>
        /// <param name="predicate">用于定义要移除的元素应满足的条件</param>
        /// <returns>移除项集合</returns>
        public List<T> Remove(Func<T, bool> predicate)
        {
            if (predicate == null)
            {
                throw new ArgumentNullException(nameof(predicate));
            }

            this.ThrowObjectDisposedException();

            var removeItems = new List<T>();
            lock (this.SyncRoot)
            {
                if (this._queue.Any(predicate))
                {
                    T[] array = this._queue.ToArray();
                    this.PrimitiveClear();

                    foreach (var item in array)
                    {
                        if (predicate(item))
                        {
                            removeItems.Add(item);
                        }
                        else
                        {
                            this._queue.Enqueue(item);
                        }
                    }
                }
            }

            return removeItems;
        }

        /// <summary>
        /// 获取队列中包含的元素数
        /// </summary>
        public int Count
        {
            get
            {
                return this._queue.Count;
            }
        }

        /// <summary>
        /// 将队列中存储的元素复制到新数组中
        /// </summary>
        /// <returns>新数组</returns>
        public T[] ToArray()
        {
            lock (this.SyncRoot)
            {
                return this._queue.ToArray();
            }
        }

        /// <summary>
        /// 清空队列,必须在停止时执行,否则后果未知
        /// </summary>
        public void Clear()
        {
            lock (this.SyncRoot)
            {
                this.PrimitiveClear();
            }
        }


        private void PrimitiveClear()
        {
            this._queue.Clear();
        }

        /// <summary>
        /// 释放资源
        /// </summary>
        public void Dispose()
        {
            if (this._disposed)
            {
                return;
            }
            this._disposed = true;

            this.Stop(false, 5000);

            this._thread.Dispose();
            this._emptyQueueWaitEventHandle.Dispose();
        }
    }

}
